package com.atguigu.flink.chapter07;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * TODO 水位超过5，侧输出流告警
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 14:06
 */
public class Flink24_State_KeyedDEMO {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 8888)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {   // 指定如何从数据中提取事件时间
                                    @Override
                                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                        return element.getTs() * 1000L; // 转换成 毫秒
                                    }
                                })
                );


        SingleOutputStreamOperator<String> resultDS = sensorDS
                .keyBy(sensor -> sensor.getId())
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    // TODO 1.定义状态
                    ValueState<Integer> valueState;
                    ValueState<List<Integer>> valueListState;
                    ListState<WaterSensor> listState;
                    MapState<String, Long> mapState;

                    // TODO 2.在open方法里，初始化状态
                    @Override
                    public void open(Configuration parameters) throws Exception {
//                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("valueState",Types.INT,0));
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("valueStateDemo", Types.INT));
                        valueListState = getRuntimeContext().getState(new ValueStateDescriptor<List<Integer>>("valueListStateDemo", Types.LIST(Types.INT)));
                        listState = getRuntimeContext().getListState(new ListStateDescriptor<WaterSensor>("listStateDemo", Types.POJO(WaterSensor.class)));
                        // mapstate默认值就是null，不能指定
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("mapStateDemo", Types.STRING, Types.LONG));
                    }

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
//                        valueState.value(); // 获取状态存储的值
//                        valueState.update(1); // 更新状态值
//                        valueState.clear(); //清空状态，注意，清空的是本组的状态，value是哪个组，就清空哪个组的状态
//
//                        Iterable<WaterSensor> waterSensors = listState.get(); // 获取状态里的数据，返回一个可迭代类型
//                        listState.add(new WaterSensor("sensor_1", 1L, 1)); // 添加单个值
//                        listState.addAll(); // 添加整个List
//                        listState.update(); // 更新整个List
//                        listState.clear();  // 清空本组状态
//
//
//                        Iterable<Map.Entry<String, Long>> entries = mapState.entries(); // 获取整个map的数据，返回一个可迭代类型
//                        mapState.get(); // 通过某个key，获取对应的value
//                        mapState.put(, ); // 添加一对 key-value
//                        mapState.putAll(); // 添加整个map
//                        mapState.contains(); // 是否包含某个 key
//                        mapState.remove(); // 删除指定key对应的数据
//                        mapState.clear();   // 清空本组的状态

                    }
                });


        resultDS.print("result");

        env.execute();
    }

}
/*
    键控状态： keyby之后
          一个分组对应一个状态，分组之间是隔离的，跟并行度无关
 */